package com.atbeijing.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * UpdateStateByKey原语用于记录历史记录，有时，我们需要在DStream中跨批次维护状态(例如流计算中累加wordcount)。针对这种情况，updateStateByKey()为我们提供了对一个状态变量的访问，用于键值对形式的DStream。给定一个由(键，事件)对构成的 DStream，并传递一个指定如何根据新的事件更新每个键对应状态的函数，它可以构建出一个新的 DStream，其内部数据为(键，状态) 对。
 * updateStateByKey() 的结果会是一个新的DStream，其内部的RDD 序列是由每个时间区间对应的(键，状态)对组成的。
 * updateStateByKey操作使得我们可以在用新信息进行更新时保持任意的状态。为使用这个功能，需要做下面两步：
 * 1. 定义状态，状态可以是一个任意的数据类型。
 * 2. 定义状态更新函数，用此函数阐明如何使用之前的状态和来自输入流的新值对状态进行更新。
 * 使用updateStateByKey需要对检查点目录进行配置，会使用检查点来保存状态。
 * 更新版的wordcount
 */
object SparkStreaming09_CP {

    def main(args: Array[String]): Unit = {

        // TODO 创建环境对象
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
        val ssc = StreamingContext.getOrCreate("cp", ()=> {
            val ssc = new StreamingContext(sparkConf, Seconds(3))
            ssc.checkpoint("cp")

            // 从socket中获取的数据是一行一行的字符串
            val socketDS: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

            //val wordToCount = socketDS.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
            val mapDS : DStream[(String, Int)] = socketDS.flatMap(_.split(" ")).map((_,1))

            // updateStateByKey方法需要传递参数，这个参数的类型为函数类型
            // 函数类型中输入有两个参数
            //    seq : Seq[Int], 将相同的key的value放置在seq集合中
            //    opt : Option[Int]

            // The checkpoint directory has not been set. Please set it by StreamingContext.checkpoint().
            // 不推荐使用
            // Redis
            //相同key的value放入seq,之所以返回Option类型是因为可能为null
            val stateDS: DStream[(String, Int)] = mapDS.updateStateByKey(
                (seq: Seq[Int], buffer: Option[Int]) => {
                    val oldCnt = buffer.getOrElse(0)
                    val newCnt = oldCnt + seq.sum
                    Option(newCnt)
                }
            )

            stateDS.print()
            ssc
        })

        ssc.start()
        ssc.awaitTermination()
    }
}
